package com.ww.exercise.coroutine.mid.q13

import kotlinx.coroutines.*
import kotlinx.coroutines.channels.Channel
import kotlinx.coroutines.flow.*
import kotlinx.coroutines.sync.Mutex
import kotlinx.coroutines.sync.withLock
import kotlin.random.Random
import kotlin.system.measureTimeMillis
import kotlin.time.Duration.Companion.seconds

// 1.数据模型定义
data class KafkaMessage(
    val id: String,
    val content: String,
    val offset: Long // 消息偏移量，用于提交
)

// 控制信号量：暂停/恢复生产者
enum class ControlSignal { PAUSE, RESUME }

// 统计指标
data class SyncMetrics(
    val pendingCount: Int, // 待处理消息数（堆积量）
    val processedPerSecond: Double, // 每秒处理速率
    val producedPerSecond: Double, // 每秒生产速率
    val lastError: String? // 最近错误信息
)

// 2.数据同步系统实现
class DataSyncSystem(
    private val scope: CoroutineScope,
    private val batchCommitSize: Int = 100 // 批量提交阈值
) {
    // 消息通道：使用buffer控制背压，超出容量时生产者会暂停
    private val messageFlow = MutableSharedFlow<KafkaMessage>(
        replay = 0,
        extraBufferCapacity = 500 // 缓冲容量，超过则触发背压
    )

    // 控制信号通道：用于传递暂停/恢复信号
    private val controlChannel = Channel<ControlSignal>(Channel.CONFLATED)

    // 状态管理（线程安全）
    private val mutex = Mutex()
    private var isPaused = false
    private var pendingCount = 0 // 堆积消息数
    private var processedCount = 0 // 总处理数
    private var producedCount = 0 // 总生产数
    private var lastError: String? = null
    private var lastCommitOffset = -1L // 最后提交的偏移量

    // 统计指标流（供外部观察）
    private val _metrics = MutableStateFlow(SyncMetrics(0, 0.0, 0.0, null))
    val metrics: StateFlow<SyncMetrics> = _metrics.asStateFlow()

    // 3.生产者：从Kafka消费消息（模拟）
    fun startProducer() {
        scope.launch(Dispatchers.IO) {
            var offset = 0L
            val produceInterval = 1000 / 1000L // 模拟每秒1000条

            while (isActive) {
                // 检查控制信号
                when (controlChannel.tryReceive().getOrNull()) {
                    ControlSignal.PAUSE -> {
                        mutex.withLock { isPaused = true }
                        println("\uD83D\uDCCC 生产者已暂停")
                    }

                    ControlSignal.RESUME -> {
                        mutex.withLock { isPaused = false }
                        println("\uD83D\uDCCC 生产者已恢复")
                    }

                    null -> Unit // 无信号，继续
                }

                // 若处于暂停状态或背压生效，等待
                if (mutex.withLock { isPaused }) {
                    delay(100)
                    continue
                }

                // 生产消息（模拟Kafka消费）
                val message = KafkaMessage(
                    id = "MSG-${offset}",
                    content = "内容-${offset}",
                    offset = offset
                )

                // 发送消息到Flow（背压点：若缓冲满则挂起）
                val sendSuccess = messageFlow.tryEmit(message)
                if (sendSuccess) {
                    mutex.withLock {
                        producedCount++
                        pendingCount++
                    }
                    offset++
                }

                // 控制生产速度（模拟每秒1000条）
                delay(produceInterval)
            }
        }
    }

    // 4.消费者：处理消息并批量提交偏移量
    fun startConsumer() {
        scope.launch(Dispatchers.Default) {
            val batchOffsets = mutableListOf<Long>()
            val processingTimes = mutableListOf<Long>() // 记录最近处理耗时

            messageFlow
                .buffer(0) // 禁用额外缓冲，直接使用上游背压
                .collect { message ->
                    try {
                        // 模拟处理耗时（约5ms/条，对应每秒200条）
                        val processTime = measureTimeMillis {
                            delay(Random.nextLong(4, 6)) // 模拟处理延迟

                            // 模拟1%的失败概率（如数据库异常）
                            if (Random.nextDouble() < 0.01) {
                                throw Exception("数据库连接超时")
                            }
                        }

                        processingTimes.add(processTime)
                        if (processingTimes.size > 100) {
                            processingTimes.removeFirst()
                        }

                        // 更新状态
                        mutex.withLock {
                            pendingCount--
                            processedCount++
                            lastError = null
                        }

                        // 收集偏移量，达到批量阈值则提交
                        batchOffsets.add(message.offset)
                        if (batchOffsets.size >= batchCommitSize) {
                            commitOffsets(batchOffsets.max())
                            batchOffsets.clear()
                        }
                    } catch (e: Exception) {
                        // 处理失败：记录错误并发送暂停信号
                        mutex.withLock {
                            lastError = e.message
                        }
                        controlChannel.send(ControlSignal.PAUSE)

                        // 等待恢复信号（最多等30秒，超时自动重试）
                        withTimeoutOrNull(30.seconds.inWholeMicroseconds) {
                            controlChannel.receive()
                        } ?: run {
                            println("⏰ 等待恢复超时，自动恢复处理")
                        }
                        batchOffsets.clear() // 清空未提交偏移量
                    }
                }
        }
    }

    // 5.提交偏移量（模拟Kafka偏移量提交）
    private suspend fun commitOffsets(maxOffset: Long) {
        // 模拟提交耗时
        delay(100)
        mutex.withLock {
            lastCommitOffset = maxOffset
        }
        println("✅ 已提交偏移量：$maxOffset")
    }

    // 6.启动指标统计（每秒更新一次）
    fun startMetricsCollector() {
        scope.launch(Dispatchers.Default) {
            var lastProcessed = 0
            var lastProduced = 0

            while (isActive) {
                delay(1000) // 每秒统计一次

                mutex.withLock {
                    // 计算速率（当前值 - 上次值）
                    val processedPerSec = (processedCount - lastProcessed).toDouble()
                    val producedPerSec = (producedCount - lastProduced).toDouble()

                    // 更新指标
                    _metrics.value = SyncMetrics(
                        pendingCount = pendingCount,
                        processedPerSecond = processedPerSec,
                        producedPerSecond = producedPerSec,
                        lastError = lastError
                    )

                    // 打印关键指标
                    println(
                        "\uD83D\uDCCA 堆积量：${pendingCount} | " +
                                "处理速率：${processedPerSec}/s | " +
                                "生产速率：${producedPerSec}/s | " +
                                "最后提交偏移量：$lastCommitOffset"
                    )

                    lastProcessed = processedCount
                    lastProduced = producedCount
                }
            }
        }
    }

    // 7.发送恢复信号（供外部调用，如故障恢复后）
    fun sendResumeSignal() {
        scope.launch {
            controlChannel.send(ControlSignal.RESUME)
        }
    }

    // 新增关闭通道，释放资源
    fun stop() {
        controlChannel.close() // 关闭控制信号通道
        (messageFlow as MutableStateFlow).resetReplayCache() // 清空Flow
    }
}

fun main() = runBlocking {
    val parentJob = Job()
    val scope = CoroutineScope(Dispatchers.Default + parentJob)
    val syncSystem = DataSyncSystem(scope)

    // 启动组件
    syncSystem.startProducer()
    syncSystem.startConsumer()
    syncSystem.startMetricsCollector()

    // 监听指标，当发生错误时，5秒后发送恢复信号
    launch {
        syncSystem.metrics.collect { metrics ->
            if (metrics.lastError != null) {
                println("❌ 检测到错误: ${metrics.lastError}，5秒后尝试恢复...")
                delay(5000)
                syncSystem.sendResumeSignal()
            }
        }
    }

    // 运行30秒后停止
    delay(30000)
    println("\uD83D\uDD04 开始停止系统...")

    syncSystem.stop()
    scope.cancel()
    parentJob.join()
    println("\uD83D\uDD1A 系统已停止")
}